package common
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client.{Admin, Connection, ConnectionFactory, Get, ResultScanner, Scan, Table}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase._
import org.apache.hadoop.hbase.filter.{BinaryComparator, CompareFilter, RegexStringComparator, RowFilter}

import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, HashMap}

class HbaseTools extends Serializable {

  private var con:Connection = null
  private val hbase_zq:String = "10.120.65.181,10.120.67.164,10.120.64.243,10.120.64.144"
  private val hbase_zqPort:String = "2181"
  private val hBase_zParent:String = "/hbase"
  def initConnection()={
    if(con==null||con.isClosed){
      val conf:Configuration = HBaseConfiguration.create()
      conf.set("hbase.zookeeper.quorum", this.hbase_zq)
      conf.set("hbase.zookeeper.property.clientPort", this.hbase_zqPort)
      conf.set("zookeeper.znode.parent", this.hBase_zParent) //some condition you should add;
      val connection = ConnectionFactory.createConnection(conf)
      con = connection
    }
  }
  def getConnection():Connection ={
    val conf:Configuration = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", this.hbase_zq)
    conf.set("hbase.zookeeper.property.clientPort", this.hbase_zqPort)
    conf.set("zookeeper.znode.parent", this.hBase_zParent) //some condition you should add;
    val connection = ConnectionFactory.createConnection(conf)
    connection
  }
  def createTable(connection:Connection,tablenameString:String,columnFamily:String)={
    val admin:Admin = connection.getAdmin
    val tableName:TableName = TableName.valueOf(tablenameString)
    val table:HTableDescriptor = new HTableDescriptor(tableName)
    val family:HColumnDescriptor = new HColumnDescriptor(columnFamily)
    table.addFamily(family)
    if(!admin.tableExists(tableName)){
      admin.createTable(table)
    }else{
      val des:HTableDescriptor = admin.getTableDescriptor(tableName)
      val families:Array[HColumnDescriptor] = des.getColumnFamilies
      val lf:Array[String] = families.map(hc=>{
        hc.getNameAsString
      })
      if(!lf.contains(tablenameString)){
        val myfamily:HColumnDescriptor = new HColumnDescriptor(columnFamily)
        admin.modifyColumn(tableName,myfamily)
      }
    }
  }

  def addFamily(connection:Connection,tablenameString:String,familynameString:String):Boolean={
    val admin:Admin = connection.getAdmin
    val Tname:TableName = TableName.valueOf(tablenameString)
    if(admin.tableExists(Tname)){
      val des:HTableDescriptor = admin.getTableDescriptor(Tname)
      val descriptor = new HColumnDescriptor(familynameString)
      des.addFamily(descriptor)
      admin.disableTable(Tname)
      admin.modifyTable(Tname,des)
      admin.enableTable(Tname)
    }
    admin.close()

    true
  }

  def addFamilys(connection:Connection,tablenameString:String,familynameStrings:Array[String]):Boolean={
    val admin:Admin = connection.getAdmin
    val Tname:TableName = TableName.valueOf(tablenameString)
    if(admin.tableExists(Tname)){
      val des:HTableDescriptor = admin.getTableDescriptor(Tname)
      val families = des.getColumnFamilies
      val familiesString:Array[String] = families.map(family=>{
        family.getNameAsString
      })
      for(familynamestring<-familynameStrings){
        if(!familiesString.contains(familynamestring)){
          val descriptor = new HColumnDescriptor(familynamestring)
          des.addFamily(descriptor)
        }
      }
      admin.disableTable(Tname)
      admin.modifyTable(Tname,des)
      admin.enableTable(Tname)
    }
    admin.close()

    true
  }


  //参数(connection,tableName="BaseMapInfos",rowkey="public",columnFamily="public_Time")
  //get"BaseMapInfos","public"
  def getFamily_column_value(connection:Connection,tableName: String, rowkey: String, columnFamily: String):mutable.HashMap[String, String] = {
    val tName: TableName = TableName.valueOf(tableName)
    val table: Table = connection.getTable(tName)
    val get: Get = new Get(Bytes.toBytes(rowkey))
    get.addFamily(Bytes.toBytes(columnFamily))
    val result: Array[Cell] = table.get(get).rawCells()
    val values: HashMap[String, String] = new HashMap[String, String]()
    if (result != null) {
      for (cell <- result) {
        val column_name: String = Bytes.toString(CellUtil.cloneQualifier(cell))
        val value: String = Bytes.toString(CellUtil.cloneValue(cell))
        values += (column_name -> value)
      }
    }
    table.close()
    values
  }

  def getFamilys_list(connection:Connection,tableName: String, rowkey: String): Array[String] = {
    //get one table's all columnfamily
    val tName = TableName.valueOf(tableName)
    val table = connection.getTable(tName)
    val htd: HTableDescriptor = table.getTableDescriptor()
    val familysDesc: Array[HColumnDescriptor] = htd.getColumnFamilies

    var familys = new ArrayBuffer[String]()
    for (hd: HColumnDescriptor <- familysDesc) {
      familys += (hd.getNameAsString)
    }
    table.close()
    familys.toArray
  }
  def makeFilter(connection: Connection,tableName:String):Array[String]={
    import  scala.collection.JavaConverters._
    val strings = Array("")
    val scan = new Scan()
    val comparator = new BinaryComparator(Bytes.toBytes(1))
    val comparator1 = new RegexStringComparator("*")
    val filter = new RowFilter(CompareFilter.CompareOp.LESS,comparator)
    scan.setFilter(filter)
    val tName = TableName.valueOf(tableName)
    val table = connection.getTable(tName)
    val scanner:ResultScanner = table.getScanner(scan)
    val res  =scanner.asScala
    for(re<-res){
      val cells = re.listCells();

    }

    strings

  }

  def main(args: Array[String]): Unit = {
    //
    val connection = getConnection()
    val stringToString = getFamily_column_value(connection, "BaseMapInfos", "  WTO_WTO", "map01")
    print(stringToString)
  }
}




